package com.atguigu.day7.procwindow

import com.atguigu.source.{SensorReading, SensorSource}
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.table.api.EnvironmentSettings
import org.apache.flink.table.api.scala._
import org.apache.flink.api.scala._
import org.apache.flink.table.api._
import org.apache.flink.types.Row
object TableProcTime {
  def main(args: Array[String]): Unit = {

    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val settings = EnvironmentSettings
      .newInstance()
      .useBlinkPlanner()
      .inStreamingMode()
      .build()

    val tEnv= StreamTableEnvironment.create(env,settings)
    val stream:DataStream[SensorReading] = env.addSource(new SensorSource)

    //'pt.proctime指定了处理时间是pt字段，必须放在最后
    val table: Table = tEnv.fromDataStream(stream, 'id, 'timeStamP as 'ts, 'temperature, 'pt.proctime)

    table
      .window(Tumble over 10.seconds on 'pt as 'w)
      .groupBy('id,'w)
      .select('id,'id.count)
      .toAppendStream[Row]
      .print()

    env.execute()




}}
